package HBase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.*;

/**
 * @Author lixinlei
 * @Date 2023/3/31 8:16
 */
public class Demo2_TableOperator {

    static Connection conn = null;
    static Admin admin = null;

    static {

        try {
            Configuration conf = HBaseConfiguration.create();

            conf.set(Constant.ZK_QUORUM,Constant.ZK_NODES);
            conf.set(Constant.ZK_CLIENTPORT,Constant.ZK_PORT);

            conn = ConnectionFactory.createConnection(conf);

            admin = conn.getAdmin();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

    }

    /**
     * 创建一个普通表
     * @param tbName
     * @param columnFamilyList
     *          每个列族中有多个属性，以键值对形式表示，所以一个Map表示一个列族
     *          创建一个表至少有一个列族，可以有多个列族，所以用List来盛放多个列族
     */
    public static void createTable(String tbName, List<Map<String,String>> columnFamilyList){
        createTable(tbName,null,null,0,null,columnFamilyList);
    }


    /**
     * 创建预分区表
     * @param tbName
     * @param startKey 第一个分区的 endKey
     * @param endKey    最后一个分区的 startKey
     * @param numRegions    分区的数量
     * @param columnFamilyList
     */
    public static void createTable(
            String tbName,
            String startKey,
            String endKey,
            int numRegions,
            List<Map<String,String>> columnFamilyList
    ){
        createTable(tbName,startKey,endKey,numRegions,null,columnFamilyList);
    }

    public static void createTable(
            String tbName,
            String[] splitKeys,
            List<Map<String,String>> columnFamilyList
    ){
        createTable(tbName,null,null,0,splitKeys,columnFamilyList);
    }


    private static void createTable(
            String tbName,
            String startKey,
            String endKey,
            int numRegions,
            String[] splitKeys,
            List<Map<String,String>> columnFamilyList
    ){
        try{

            //如果表不存在，可以创建
            if(!Tools.isTableExists(admin,tbName)){

                //创建一个 TableDescriptor 对象，HTableDescriptor 已经是过期类
                TableDescriptorBuilder.ModifyableTableDescriptor mtd =
                        new TableDescriptorBuilder.ModifyableTableDescriptor(TableName.valueOf(tbName));

                //判断列族的数量至少为 1
                if(null!=columnFamilyList && columnFamilyList.size()>=1){
                    //遍历所有需要添加的列族
                    for (Map<String, String> cf : columnFamilyList) {

                        //如果表示列族的Map中没有列族名，使用默认列族名进行添加
                        String cfName = "default";
                        //必须保证列族由 列族名 这个属性
                        if(cf.containsKey(Constant.COLUMNFAMILY_NAME)){
                            cfName = cf.get(Constant.COLUMNFAMILY_NAME);
                        }

                        //创建每个列族的描述对象，HColumnDescriptor 已经是过期类
                        ColumnFamilyDescriptorBuilder.ModifyableColumnFamilyDescriptor mcfd =
                                new ColumnFamilyDescriptorBuilder.ModifyableColumnFamilyDescriptor(Bytes.toBytes(cfName));


                        //@TODO 这里需要对列族的每一个属性进行判断，如果参数传递了当前列族的这个属性，需要把此属性设置到 mcfd 中
                        if(cf.containsKey(Constant.COLUMNFAMILY_MAXVERSIONS)){
                            mcfd.setMaxVersions(Integer.parseInt(cf.get(Constant.COLUMNFAMILY_MAXVERSIONS)));
                        }


                        mtd.setColumnFamily(mcfd);

                    }
                }else{
                    System.out.println("创建表至少需要一个列族");
                }

                // startKey 和 endKey 必须有值且 分区数量必须大于0 的情况 才可以创建预分区表
                if(startKey!=null && !startKey.equals("") && endKey!=null && !endKey.equals("") && numRegions>0){
                    admin.createTable(
                            mtd,
                            Bytes.toBytes(startKey),
                            Bytes.toBytes(endKey),
                            numRegions
                    );
                }else if(splitKeys!=null && splitKeys.length>0){
                    //如果有分区临界点的数组，使用二维数组的形式创建分区表
                    byte[][] splits = new byte[splitKeys.length][];

                    for(int i=0;i<splitKeys.length;i++){
                        splits[i] = Bytes.toBytes(splitKeys[i]);
                    }

                    admin.createTable(mtd,splits);
                }else{
                    //否则只能创建普通表
                    admin.createTable(mtd);
                }

            }else{
                System.out.println("当前"+tbName+"表已经存在，无需重复添加");
            }


        }catch (Exception e){
            e.printStackTrace();
        }finally {
            Tools.closed(admin,conn);
        }
    }


    /**
     * 修改或删除列族信息
     * @param tbName
     * @param columnFamilies
     *          如果参数是 String 类型的不定长，表示的是删除
     *          如果参数是 Map 类型的不定长，表示的是添加或修改
     */
    public static void alterTable(String tbName,Object... columnFamilies){

        try{
            if(Tools.isTableExists(admin,tbName)){

                TableDescriptorBuilder.ModifyableTableDescriptor descriptor =
                        (TableDescriptorBuilder.ModifyableTableDescriptor)admin.getDescriptor(TableName.valueOf(tbName));

                //遍历参数
                //TODO 暂时没有想到更好的方式
                for (Object columnFamily : columnFamilies) {
                    //如果参数的泛型是String，做删除列族操作
                    if(columnFamily instanceof String){
                        String cf = (String)columnFamily;
                        if(descriptor.hasColumnFamily(Bytes.toBytes(cf))){
                            descriptor.removeColumnFamily(Bytes.toBytes(cf));
                        }else{
                            System.out.println("当前"+cf+"列族不存在");
                        }
                    }else if(columnFamily instanceof Map){
                        //如果参数的泛型是 Map，做添加或修改列族操作
                        Map<String,String> cfMap = (Map<String,String>)columnFamily;

                        //首先保证列族必须有名字
                        if(cfMap.containsKey(Constant.COLUMNFAMILY_NAME)){
                            //获取列族名
                            String cfName = cfMap.get(Constant.COLUMNFAMILY_NAME);

                            //如果此列族已经存在于表中，就是修改
                            if(descriptor.hasColumnFamily(Bytes.toBytes(cfName))){

                                //因为是修改，所以先获取原有的列族信息，在原列族基础上进行修改
                                ColumnFamilyDescriptorBuilder.ModifyableColumnFamilyDescriptor mcfd =
                                        (ColumnFamilyDescriptorBuilder.ModifyableColumnFamilyDescriptor)descriptor.getColumnFamily(Bytes.toBytes(cfName));

                                if(cfMap.containsKey(Constant.COLUMNFAMILY_MAXVERSIONS)){
                                    mcfd.setMaxVersions(Integer.parseInt(cfMap.get(Constant.COLUMNFAMILY_MAXVERSIONS)));
                                }

                                descriptor.modifyColumnFamily(mcfd);
                            }else{
                                //如果此列族不存在于表中，就是添加
                                //因为是添加，所以需要创建一个新的列族描述对象
                                ColumnFamilyDescriptorBuilder.ModifyableColumnFamilyDescriptor mcfd =
                                        new ColumnFamilyDescriptorBuilder.ModifyableColumnFamilyDescriptor(Bytes.toBytes(cfName));

                                if(cfMap.containsKey(Constant.COLUMNFAMILY_MAXVERSIONS)){
                                    mcfd.setMaxVersions(Integer.parseInt(cfMap.get(Constant.COLUMNFAMILY_MAXVERSIONS)));
                                }
                                descriptor.setColumnFamily(mcfd);
                            }
                        }
                    }
                }
                admin.modifyTable(descriptor);
            }else{
                System.out.println("当前"+tbName+"表不存在，无法修改");
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            Tools.closed(admin,conn);
        }
    }


    /**
     * 修改表状态，默认关闭连接
     * @param tbName
     * @param flag
     */
    public static void alterTableStatus(String tbName,boolean flag){
        alterTableStatus(tbName,flag,true);
    }

    /**
     * 修改表状态，可以不关闭连接，当做工具使用
     * @param tbName
     * @param flag true：启用   false：禁用
     * @param closeFlag true：关闭连接  false：不关闭连接
     */
    public static void alterTableStatus(String tbName,boolean flag,boolean closeFlag){
        try{
            //下面多次使用，提出公共的
            TableName table = TableName.valueOf(tbName);

            //确保表存在
            if(Tools.isTableExists(admin,tbName)){
                //如果表为禁用状态，且想要启用，执行启用
                if (flag && admin.isTableDisabled(table)){
                    admin.enableTable(table);
                }else if(!flag && admin.isTableEnabled(table)){
                    //如果表为可用状态，且想要禁用，执行禁用
                    admin.disableTable(table);
                }else{
                    System.out.println("状态无需改变");
                }

            }else{
                System.out.println("当前"+tbName+"表不存在，无法修改");
            }

        }catch (Exception e){
            e.printStackTrace();
        }finally {
            if(closeFlag){
                Tools.closed(admin,conn);
            }
        }
    }

    /**
     * 获取单表信息
     * @param tbName
     */
    public static void getTableInfo(String tbName){
        String[] tbNames = {tbName};
        getTableInfo(tbNames);
    }

    /**
     * 获取多表信息
     * @param tbNames
     */
    public static void getTableInfo(String[] tbNames){

        try{

            //遍历所有要查询的表
            for (String tbName : tbNames) {

                //确保表存在
                if(Tools.isTableExists(admin,tbName)){
                    //获取当前表的表描述信息
                    TableDescriptor descriptor = admin.getDescriptor(TableName.valueOf(tbName));

                    //从表描述信息中获取相应信息
                    System.out.println("tableName:"+descriptor.getTableName());

                    //获取当前表的所有的列族
                    ColumnFamilyDescriptor[] columnFamilies = descriptor.getColumnFamilies();

                    //把当前表的所有列族的所有属性进行遍历输出
                    //TODO 根据需要获取不同的列族属性
                    for (ColumnFamilyDescriptor columnFamily : columnFamilies) {
                        System.out.println("name:"+columnFamily.getNameAsString());
                        System.out.println("maxVersions:"+columnFamily.getMaxVersions());
                        System.out.println("======================================================");
                    }
                }

                System.out.println("========================================================");

            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            Tools.closed(admin,conn);
        }

    }

    /**
     * 删除表
     * @param tbNames
     */
    public static void dropTable(String... tbNames){

        try{
            //不定长可以是0个参数，所以必须至少有一个表才可以删除
            if(tbNames.length>0){
                for (String tbName : tbNames) {
                    if(Tools.isTableExists(admin,tbName)){
                        //删除表之前必须先禁用表
                        alterTableStatus(tbName,false,false);
                        admin.deleteTable(TableName.valueOf(tbName));
                    }else{
                        System.out.println("当前"+tbName+"表不存在，无法删除");
                    }
                }
            }else{
                System.out.println("没有要删除的表");
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            Tools.closed(admin,conn);
        }

    }

    public static void main(String[] args) {

        //1.创建普通表
//        List<Map<String,String>> cfList = new ArrayList<>();
//
//        Map<String,String> cfMap1 = new HashMap<>();
//        cfMap1.put("name","baseInfo");
//        cfMap1.put("versions","3");
//
//        Map<String,String> cfMap2 = new HashMap<>();
//        cfMap2.put("name","moreInfo");
//        cfMap2.put("versions","5");
//
//        Map<String,String> cfMap3 = new HashMap<>();
//        cfMap3.put("name","newInfo");
//        cfMap3.put("versions","10");
//
//        cfList.add(cfMap1);
//        cfList.add(cfMap2);
//
//        createTable("hero",cfList);

        //2.创建预分区表
//        createTable(
//                "splitTable",
//                "001",
//                "010",
//                3,
//                cfList
//        );

        //3.创建预分区表
//        String[] splits = {"001","010","020","030"};
//        createTable(
//                "ztq",
//                splits,
//                cfList
//        );

        //4.修改表的列族信息
//        alterTable("hero",cfMap1,cfMap2,cfMap3);
//        alterTable("hero","newInfo");

        //5.启用禁用表
//        alterTableStatus("hero",false);

        //6.获取表信息
//        String[] tables = {"hero","student"};
//        getTableInfo(tables);

        //7.删除表
        dropTable("ztq","hero","splitTable");
    }

}